package com.example.common.config.rabbit.common;

import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.example.common.constant.MessageConstant;
import com.example.common.constant.RabbitContent;
import com.example.common.web.service.MqMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * <p>
 * <p>
 * MQ 可靠消息如何实现：
 * 一、消息重复消费：
 * 1. 消费者消费方法设计成幂等，只要消息消费了并且修改了某个状态，那么以后的消息则不会再处理
 * 2. 使用 message.getMessageProperties().getRedelivered() 来判断是否是重发的消息，再结合业务进行判断
 * 二、消息积压：
 * 1. 如果由于消息生产者生产消息效率大于消费者消费能力，那么可以增家消费者数量
 * 三、 消息丢失：
 * 1. 消息丢失，做好 MQ 的确认机制，如果一旦发送失败的消息，将失败的消息存入数据库中进行持久化，做数据库的定期扫描重试消息
 * <p>
 * RabbitTemplate 包装, 用来减少代码，将每次发送的消息保存到数据库中
 * 作用是完成数据库的保存
 * </p>
 *
 * @author zyred
 * @since v 0.1
 **/
@Slf4j
@Configuration
@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitTemplateWrapper {

    @Autowired private MqMessageService mqMessageService;
    @Autowired private RabbitProperties rabbitProperties;
    @Autowired private RabbitTemplate rabbitTemplate;
    @Autowired private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 发送与保存
     *
     * @param exchange   交换机名称
     * @param routingKey 路由键
     * @param message    发送的消息
     */
    public void sendAndSaveAsync(String exchange, String routingKey, Object message) {

        // 将此ID作为发送到 MQ 的 tag
        final String id = String.valueOf(IdWorker.getId());

        // 1. 保存 mq 消息到数据库中
        CompletableFuture.runAsync(
                () -> this.mqMessageService.insertMessage(id, exchange,
                        routingKey, message.getClass().getName(), message),
                this.threadPoolExecutor
        );

        // 创建 MQ 唯一标识
        CorrelationData data = new CorrelationData();
        data.setId(id);

        // 2. 发送消息到服务器
        this.rabbitTemplate.convertAndSend(exchange, routingKey, message, data);
    }


    /**
     * 消息一旦发生意外或成功都将被修改到数据中
     *
     * @param status        消息状态 {@see com.example.common.constant.MessageConstant}
     * @param reason        消息错误原因
     * @param id            消息 MQ ID
     */
    public void alertStatusAsync(int status, String reason, String id) {

        CompletableFuture.runAsync(
                () -> {
                    this.mqMessageService.updateMessage(status, reason, id);
                }, threadPoolExecutor
        );
    }

    @Bean
    public MessageConverter messageConverter() {
        log.info("Enable jackson2 message converter.");
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 初始化为 rabbitTemplate 设置确认机制
     */
    @PostConstruct
    public void initSetting() {

        // 开启了 broker 消息确认才执行
        if (this.rabbitProperties.isPublisherConfirms()) {
            this.rabbitTemplate.setConfirmCallback(
                    /*
                     * correlationData: 消息唯一关联数据
                     * ack:             消息是否成功收到
                     * cause:           错误发送的原因
                     */
                    (correlationData, ack, cause) -> {
                        log.info(">>>>>>>>>>>>>>>>>>>Broker>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
                        if (!ack) {
                            log.info("唯一关联数据：{}", correlationData);
                            log.info("错误发送的原因：{}", cause);
                        }
                        log.info("消息是否成功收到：{}", ack);
                        log.info("<<<<<<<<<<<<<<<<<<<Broker<<<<<<<<<<<<<<<<<<<<<<<<<<<<<");
                    }
            );
            log.info("Rabbit publisher confirms enable.");
        } else {
            log.info("Rabbit publisher confirms disable.");
        }

        // 未开启消息抵达通知
        if (!this.rabbitProperties.isPublisherReturns()) {
            log.info("Rabbit publisher returns disable.");
            return;
        }

        // 发送消息成功后是不会触发这里的
        this.rabbitTemplate.setReturnCallback(
                /*
                 * message:     投递失败的消息详情
                 * replyCode:   回复的状态码
                 * replyText:   回复的文本消息
                 * exchange:    使用的交换机名称
                 * routingKey:  使用的路由键名称
                 */

                /*

                消息详情：[]
                状态码：312
                错误消息：NO_ROUTE
                交换机名称：order-event-exchange
                路由键名称：order.create.order.test
                 */
                (message, replyCode, replyText, exchange, routingKey) -> {

                    log.info(">>>>>>>>>>>>>>>>>>>>>Queue>>>>>>>>>>>>>>>>>>>>>>>>>>>");
                    log.info("消息详情：{}", message);
                    log.info("状态码：{}", replyCode);
                    log.info("错误消息：{}", replyText);
                    log.info("交换机名称：{}", exchange);
                    log.info("路由键名称：{}", routingKey);
                    log.info("<<<<<<<<<<<<<<<<<<<<<<Queue<<<<<<<<<<<<<<<<<<<<<<<<<<");

                    // 只要这里一旦触发，那么一定是持久化失败
                    int status = MessageConstant.MessageStatus.PERSISTENCE_ERROR.getCode();
                    // 组装错误原因
                    String reason = "状态码:(" + replyCode + ") 原因:(" + replyText + ")";
                    // 从消息里面拿到错误 ID
                    String id = (String) message.getMessageProperties().getHeaders().get(RabbitContent.RABBIT_KEY);
                    this.alertStatusAsync(status, reason, id);
                }
        );

        log.info("Rabbit publisher returns enable.");
    }
}

